package lamma.lab2.job;

import java.lang.reflect.Field;
import java.lang.reflect.InvocationTargetException;
import java.lang.reflect.Method;
import java.util.Arrays;
import java.util.Collection;
import java.util.Collections;
import java.util.HashMap;
import java.util.HashSet;
import java.util.Iterator;
import java.util.LinkedList;
import java.util.List;
import java.util.Map;
import java.util.Set;

import lamma.helper.ReflectionHelper;

import com.google.common.collect.ConcurrentHashMultiset;
import com.google.common.collect.Multiset;

/**
 * @author maxcellent
 *
 */
public class JobExecutor<T> {
	
	/**
	 * @author maxcellent
	 *	
	 */
	private class _Job {
		
		/**
		 * job object
		 */
		Object jobObj;
		
		// status
		boolean finished;

		// ========================values derived from jobObj========================
		
		// ===================fields running status==================
		/**
		 * lists of ready fields
		 */
		Map<String, _FieldStatus> fieldStatusMap;
		
		// ============steps running status==================
		Map<String, _Step> steps;
		
		/**
		 * field ready listners
		 */
		Map<String, List<_FieldReadyListener>> fieldReadyListner;
		
		public _Job(Object jobObj) {
			this.jobObj = jobObj;
			
			this.finished = false;
			
			// init step map
			this.steps = new HashMap<String, JobExecutor<T>._Step>();
			
			Method [] methods = jobObj.getClass().getDeclaredMethods();
			for (Method method : methods) {
				Step step = method.getAnnotation(Step.class);
				
				if (step != null) {
					// init all steps into pending steps
					this.steps.put(method.getName(), new _Step());
				}
			}
			
			// init fields map
			this.fieldStatusMap = new HashMap<String, JobExecutor._FieldStatus>();
			
			Field [] fields = jobObj.getClass().getDeclaredFields();
			for (Field field : fields) {
				ProvidedBy providedBy = field.getAnnotation(ProvidedBy.class);
				AfterStep afterStep = field.getAnnotation(AfterStep.class);
				
				if (providedBy == null && afterStep == null) {
					fieldStatusMap.put(field.getName(), _FieldStatus.READY);
				} else {
					fieldStatusMap.put(field.getName(), _FieldStatus.PENDING);
				}
				
				if (afterStep != null) {
					_Step step = steps.get(afterStep.value());
					step.finishFields.add(field.getName());
				}
			}
			
			// init listener
			fieldReadyListner = new HashMap<String, List<_FieldReadyListener>>();
		}

		/**
		 * @throws SecurityException
		 * @throws IllegalArgumentException
		 * @throws NoSuchMethodException
		 * @throws IllegalAccessException
		 * @throws InvocationTargetException
		 * @throws NoSuchFieldException
		 * @throws InstantiationException
		 */
		public void checkAndRun() throws SecurityException, IllegalArgumentException, NoSuchMethodException, IllegalAccessException, InvocationTargetException, NoSuchFieldException, InstantiationException {
			checkAndProcessFields();
			checkAndProcessSteps();
			checkFinish();
		}

		/**
		 * @throws SecurityException
		 * @throws NoSuchFieldException
		 * @throws IllegalArgumentException
		 * @throws IllegalAccessException
		 * @throws InstantiationException
		 * @throws InvocationTargetException
		 * @throws NoSuchMethodException 
		 */
		private void checkAndProcessFields() throws SecurityException, NoSuchFieldException, IllegalArgumentException, IllegalAccessException, InstantiationException, InvocationTargetException, NoSuchMethodException {
			for (final String fieldName : this.fieldStatusMap.keySet()) {
				_FieldStatus status = this.fieldStatusMap.get(fieldName);
				
				if (status == _FieldStatus.PENDING) {
					// may be pending for 
					Field field = this.jobObj.getClass().getDeclaredField(fieldName);
					
					final ProvidedBy providedBy = field.getAnnotation(ProvidedBy.class);
					
					if (providedBy == null) {	// if it's pending for steps, nothing to do here
						continue;
					}
					
					if (this.allFieldsReady(providedBy.params())) {	// return ture even params are zero length array
						// kick off job
						this.fieldStatusMap.put(fieldName, _FieldStatus.PROCESSING);
						
						Object [] params = ReflectionHelper.readFields(this.jobObj, providedBy.params());
						
						processField(providedBy.job(), providedBy.result(), new _FieldReadyListener() {
							@Override
							public void onFieldReady(Object val) throws IllegalArgumentException, SecurityException, IllegalAccessException, NoSuchFieldException {
								ReflectionHelper.writeField(jobObj, fieldName, val);
								fieldStatusMap.put(fieldName, _FieldStatus.READY);	// mark as ready
							}
						}, params);
					}		
				}
			}
		}

		/**
		 * @throws SecurityException
		 * @throws NoSuchMethodException
		 * @throws IllegalArgumentException
		 * @throws IllegalAccessException
		 * @throws InvocationTargetException
		 */
		private void checkAndProcessSteps() throws SecurityException, NoSuchMethodException, IllegalArgumentException, IllegalAccessException, InvocationTargetException {			
			// check all un-processed steps
			for (String methodName : steps.keySet()) {
				_Step step = steps.get(methodName);
				if (!step.proccessed) {
					// this will never be null
					Step stepMeta = this.jobObj.getClass().getDeclaredMethod(methodName).getAnnotation(Step.class);
					
					if (this.allFieldsReady(stepMeta.params()) && this.allStepsReady(stepMeta.steps())) {
						// process step
						step.proccessed = true;
						
						Object result = ReflectionHelper.invokeMethod(this.jobObj, methodName);
						if (result instanceof AsyncStep) {
							step.asyncStep = (AsyncStep) result;
							break;
						}
					}
				}
			}
		}
		
		/**
		 * check job or step finish
		 * 
		 * @throws IllegalArgumentException
		 * @throws SecurityException
		 * @throws IllegalAccessException
		 * @throws NoSuchFieldException
		 */
		private void checkFinish() throws IllegalArgumentException, SecurityException, IllegalAccessException, NoSuchFieldException {			
			// check steps
			boolean allStepsFinished = true;
			for (_Step step : this.steps.values()) {
				if (step.finished()) {
					if (step.finishFields != null) {
						// set all these fields ready
						for (String fieldName : step.finishFields) {
							this.fieldStatusMap.put(fieldName, _FieldStatus.READY);
						}
						// set to empty list so that field will only be set once
						step.finishFields = null;	
					}
				} else {
					allStepsFinished = false;
				}
			}
			
			// check fields
			boolean allFieldsReady = true;
			for (String fieldName : this.fieldStatusMap.keySet()) {
				_FieldStatus status = this.fieldStatusMap.get(fieldName);
				if (status == _FieldStatus.READY) {
					List<_FieldReadyListener> listeners = this.fieldReadyListner.get(fieldName);
					
					if (listeners != null) {
						Object val = ReflectionHelper.readField(this.jobObj, fieldName);
						for (_FieldReadyListener listener : listeners) {
							listener.onFieldReady(val);
						}
						this.fieldReadyListner.remove(fieldName);
					}
				} else {
					allFieldsReady = false;
				}
			}
			if (!allFieldsReady || !allStepsFinished) {
				return;
			}
			
			// all finished, callback listeners
			this.finished = true;
			
			// call field ready listners
			for (String fieldName : this.fieldReadyListner.keySet()) {				
				List<_FieldReadyListener> listeners = fieldReadyListner.get(fieldName);
				Object val = ReflectionHelper.readField(this.jobObj, fieldName);
				for (_FieldReadyListener listener : listeners) {
					listener.onFieldReady(val);
				}
			}
			this.fieldReadyListner.clear();
		}
		
		private boolean allFieldsReady(String ...fieldNamePaths) {
			for (String fieldNamePath : fieldNamePaths) {
				String fieldName = ProvidedBy.Resolver.resolveRoot(fieldNamePath);
				if (this.fieldStatusMap.get(fieldName) != _FieldStatus.READY) {
					return false;
				}
			}
			return true;
		}
		
		private boolean allStepsReady(String ...stepNames) {
			for (String stepName : stepNames) {
				if (!this.steps.get(stepName).finished()) {
					return false;
				}
			}
			return true;
		}

		public void listenField(String fieldName, _FieldReadyListener listener) throws IllegalArgumentException, SecurityException, IllegalAccessException, NoSuchFieldException {
			if (this.fieldStatusMap.get(fieldName) == _FieldStatus.READY) {
				listener.onFieldReady(ReflectionHelper.readField(jobObj, fieldName));
			} else {
				List<_FieldReadyListener> listeners = this.fieldReadyListner.get(fieldName);
				if (listeners == null) {
					listeners = new LinkedList<JobExecutor._FieldReadyListener>();
					this.fieldReadyListner.put(fieldName, listeners);
				}
				listeners.add(listener);
			}
		}
	}
	
	/**
	 * @author maxcellent
	 *
	 *	used to identify an reusable job
	 */
	private class _JobKey {
		
		String className;
		
		Object [] args;

		_JobKey(Class<?> clazz, Object[] args) {
			super();
			this.className = clazz.getName();
			this.args = args;
		}

		@Override
		public String toString() {
			return "_JobKey [className=" + className + ", args="
					+ Arrays.toString(args) + "]";
		}

		@Override
		public int hashCode() {
			final int prime = 31;
			int result = 1;
			result = prime * result + getOuterType().hashCode();
			result = prime * result + Arrays.hashCode(args);
			result = prime * result
					+ ((className == null) ? 0 : className.hashCode());
			return result;
		}

		@Override
		public boolean equals(Object obj) {
			if (this == obj)
				return true;
			if (obj == null)
				return false;
			if (getClass() != obj.getClass())
				return false;
			@SuppressWarnings("unchecked")
			_JobKey other = (_JobKey) obj;
			if (!getOuterType().equals(other.getOuterType()))
				return false;
			if (!Arrays.equals(args, other.args))
				return false;
			if (className == null) {
				if (other.className != null)
					return false;
			} else if (!className.equals(other.className))
				return false;
			return true;
		}

		private JobExecutor<T> getOuterType() {
			return JobExecutor.this;
		}
	}
	
	private interface _FieldReadyListener {
		void onFieldReady(Object val) throws IllegalArgumentException, SecurityException, IllegalAccessException, NoSuchFieldException;
	}
	
	private class _Step {
		
		boolean proccessed = false;
		
		AsyncStep asyncStep;
		
		List<String> finishFields = new LinkedList<String>();
		
		boolean finished() {
			if (this.asyncStep == null) {
				return this.proccessed;
			} else {
				return this.asyncStep.isDone();
			}
		}
	}

	/**
	 * @author maxcellent
	 *
	 */
	private static enum _FieldStatus {
		/**
		 * 
		 */
		READY,
		
		/**
		 * 
		 */
		PROCESSING,
		
		/**
		 * 
		 */
		PENDING,
	}
	
	/**
	 * @author maxcellent
	 *
	 */
	public static class BatchParam<T> {
		
		Set<List<Object>> params = new HashSet<List<Object>>();
		
		/**
		 * @param args
		 */
		public void addParam(Object ...args) {
			params.add(Arrays.asList(args));
		}
		
		/**
		 * @return
		 */
		public Collection<List<Object>> params() {
			return Collections.unmodifiableCollection(params);
		}
	}

	public static class BatchResult<T> {

		BatchParam<T> param;
		
		_FieldReadyListener listener;
		
		Map<List<Object>, T> paramsResultMap = new HashMap<List<Object>, T>();
		
		BatchResult(BatchParam<T> param, _FieldReadyListener listener) {
			this.param = param;
			this.listener = listener;
		}

		/**
		 * @param args
		 * @return
		 */
		public T lookupResult(Object ...args) {
			return paramsResultMap.get(Arrays.asList(args));
		}
		
		/**
		 * @return
		 */
		public Collection<T> results() {
			return paramsResultMap.values();
		}

		public void resultReady(List<Object> params, T val) throws IllegalArgumentException, SecurityException, IllegalAccessException, NoSuchFieldException {
			paramsResultMap.put(params, val);
			
			if (paramsResultMap.keySet().size() == param.params.size()) {	// all results are ready
				listener.onFieldReady(this);
			}
		}
	}

	private T jobObj;
	
	private Map<_JobKey, _Job> reusableJobsMap;
	
	private Multiset<_Job> runningJobs;
	
	//============used to init job============
	private Class<T> jobClass;

	private Object [] args;
	
	public JobExecutor (Class<T> jobClass, Object ...args) {
		this.jobClass = jobClass;
		this.args = args;
	}
	
	public JobExecutor (T jobObj) {
		this.jobObj = jobObj;
	}
	
	/**
	 * @param jobObj
	 * @return
	 */
	public Result<T> go() {
		try {
			this.init();
			this.start();
			return new Result<T>(this.jobObj, Result.Status.SUCCESS, null);
		} catch (InvocationTargetException e) {
			// exception thrown when the invoked method throws exception 
			return new Result<T>(this.jobObj, Result.Status.FAILED, e.getCause());
		} catch (NoSuchMethodException e) {
			e.printStackTrace();
			return new Result<T>(this.jobObj, Result.Status.FAILED, new LammaException("Please run health check", e));
		} catch (NoSuchFieldException e) {
			e.printStackTrace();
			return new Result<T>(this.jobObj, Result.Status.FAILED, new LammaException("Please run health check", e));
		} catch (IllegalArgumentException e) {
			e.printStackTrace();
			return new Result<T>(this.jobObj, Result.Status.FAILED, new LammaException("Please run health check", e));
		} catch (InstantiationException e) {
			e.printStackTrace();
			return new Result<T>(this.jobObj, Result.Status.FAILED, new LammaException("Please run health check", e));
		} catch (SecurityException e) {
			e.printStackTrace();
			return new Result<T>(this.jobObj, Result.Status.FAILED, new LammaException("Please run health check", e));
		} catch (IllegalAccessException e) {
			e.printStackTrace();
			return new Result<T>(this.jobObj, Result.Status.FAILED, new LammaException("Please run health check", e));
		} catch (Throwable e) {
			e.printStackTrace();
			return new Result<T>(this.jobObj, Result.Status.FAILED, e);
		}
	}
	
	private void init() throws IllegalArgumentException, InstantiationException, IllegalAccessException, InvocationTargetException {
		if (jobObj == null) {
			jobObj = ReflectionHelper.newInstance(jobClass, args);
		}
	}

	/**
	 * @throws IllegalArgumentException
	 * @throws IllegalAccessException
	 * @throws SecurityException
	 * @throws InstantiationException
	 * @throws NoSuchMethodException
	 * @throws InvocationTargetException
	 * @throws NoSuchFieldException
	 */
	private void start() throws IllegalArgumentException, IllegalAccessException, SecurityException, InstantiationException, NoSuchMethodException, InvocationTargetException, NoSuchFieldException {
		if (jobObj == null) {
			throw new NullPointerException("Job instance can't be null");
		}
		
		_Job mainJob = new _Job(jobObj);
		
		// job pool for reusable joe
		this.reusableJobsMap = new HashMap<_JobKey, _Job>();
		
		// init running jobs
		this.runningJobs = ConcurrentHashMultiset.create();
		this.runningJobs.add(mainJob);
		
		while (!runningJobs.isEmpty()) {
			Iterator<_Job> i = runningJobs.iterator();
			while (i.hasNext()) {
				_Job job = i.next();
				job.checkAndRun();
				if (job.finished) {
					i.remove();
				}
			}
		}
	}
	
	/**
	 * @param jobClazz
	 * @param args
	 * @return
	 * @throws IllegalArgumentException
	 * @throws InstantiationException
	 * @throws IllegalAccessException
	 * @throws InvocationTargetException
	 * @throws NoSuchFieldException 
	 * @throws SecurityException 
	 */
	@SuppressWarnings({ "rawtypes", "unchecked" })
	private void processField(Class<?> jobClazz, String fieldName, _FieldReadyListener listener, Object ...args) throws IllegalArgumentException, InstantiationException, IllegalAccessException, InvocationTargetException, SecurityException, NoSuchFieldException {
		if (args.length == 1 && args[0] instanceof BatchParam) {
			// if it's a batch param input, process in batch
			BatchParam<?> batchParam = (BatchParam<?>) args[0];
			final BatchResult batchResult = new BatchResult(batchParam, listener);
			
			for (final List<Object> params : batchParam.params) {
				processField(jobClazz, fieldName, new _FieldReadyListener() {
					@Override
					public void onFieldReady(Object val) throws IllegalArgumentException,
							SecurityException, IllegalAccessException, NoSuchFieldException {
						batchResult.resultReady(params, val);
					}
				}, params.toArray());
			}
		} else {
			// normal case			
			Reusable reusable = jobClazz.getAnnotation(Reusable.class);
			
			if (reusable == null) {
				// create a new job instance
				_Job job = new _Job(ReflectionHelper.newInstance(jobClazz, args));
				job.listenField(fieldName, listener);
				runningJobs.add(job);
			} else {
				_JobKey key = new _JobKey(jobClazz, args);
				_Job job = this.reusableJobsMap.get(key);
				if (job == null) {
					job = new _Job(ReflectionHelper.newInstance(jobClazz, args));	
					runningJobs.add(job);
					reusableJobsMap.put(key, job);
				} 
				
				job.listenField(fieldName, listener);
			}
		}
	}
}